package com.ruoyi.eqmonitor.dispose.service;

import cn.hutool.core.util.NumberUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.ruoyi.common.redis.service.RedisService;
import com.ruoyi.eqmonitor.dispose.resolve.MqttMessageResolve;
import com.ruoyi.eqmonitor.domain.BzDataEntity;
import com.ruoyi.eqmonitor.domain.BzStartStopEntity;
import com.ruoyi.eqmonitor.dtos.OptDeviceDto;
import com.ruoyi.eqmonitor.service.IBzDataService;
import com.ruoyi.eqmonitor.service.IBzStartStopService;
import com.ruoyi.eqmonitor.service.IOptDeviceService;
import com.ruoyi.eqmonitor.utils.Constant;
import com.ruoyi.eqmonitor.utils.SecretTool;
import io.lettuce.core.support.caching.RedisCache;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * 异步消费消息的实现类
 *
 * @author dh
 */
@Slf4j
@Component
public class AsyncConsumeMsg {

    @Resource
    private List<MqttMessageResolve> messageResolves;
    @Resource
    private IOptDeviceService iOptDeviceService;

    /**
     * 异步消息处理
     */
    @Async
    public void messageArrived(String topic, MqttMessage message) {
        try {
            //String json = Starwsn.messageResolve(topic, message.getPayload());
            log.info("topic:{}", topic);
            String json = new String(message.getPayload());
            if(topic.equals("data_list_bigData/bz_data_json")){
                dealBzData(json);
                return;
            }
            JSONObject jsonobject = revertToHxwl(topic, json);
            String status = jsonobject.getStr("status");

            // 状态非success的不做处理
            if (!Objects.equals(status, "success")) {
                log.error("主题:{} 【订阅消息解析失败, 原因: status 非 success】\n{}", topic, jsonobject);
                return;
            }

            // 根据报文msgType将不同类型消息单独处理
            // msgType 1 设备参数 2 监测数据 3 告警信息 只处理2和3类型的数据
            String snCode = jsonobject.getStr("snCode");
            JSONArray msg = jsonobject.getJSONArray("msg");
            int msgType = jsonobject.getInt("msgType");

            // 判断设备是否存在
            OptDeviceDto deviceInfo = iOptDeviceService.getDeviceByRealCode(snCode);
            if (deviceInfo == null) {
                log.error("snCode: {}, 消息解析失败, 设备不存在\n{}", snCode, jsonobject);
                return;
            }

            // 转换方法 目前需要转换 监测数据 液位高度 统一单位为m
            if (Constant.MqttMsgType.MONITOR_DATA.getCode() == msgType) {
                msg.parallelStream().forEach(item -> {
                    JSONObject jsonObject = (JSONObject) item;
                    String dataUnit = jsonObject.getStr("dataUnit", null);
                    String monitorCode = jsonObject.getStr("monitorCode");
                    if (Constant.MonitorInfo.M_0x05D1.getCode().equals(monitorCode) && "cm".equals(dataUnit)) {
                        Double monitorData = jsonObject.getDouble("monitorData", 0D);
                        double conversionData = NumberUtil.div(monitorData.doubleValue(), 100.0, 3);
                        jsonObject.set("monitorData", conversionData);
                        jsonObject.set("dataUnit", Constant.MonitorInfo.M_0x05D1.getUnit());
                        log.info("设备:{} , 原有液位高度: {}{}, 转换后液位高度: {}{}", snCode, monitorData, dataUnit, conversionData, Constant.MonitorInfo.M_0x05D1.getUnit());
                    }
                    //排水监测为负数 调整为0
                    if(Constant.MonitorInfo.M_0x05D1.getCode().equals(monitorCode) && deviceInfo.getDeviceType().equals("SR500-N-D2-LD")){
                        Double monitorData = jsonObject.getDouble("monitorData", 0D);
                        if(monitorData < 0){
                            log.info("积水监测设备：{}上报数据为{}调整为0", snCode, monitorData);
                            jsonObject.set("monitorData", 0);
                        }
                    }
                });
            }
            /*
             * TODO 这里可以定义一些实现MqttMessageResolve的：[热点缓存、设备在线状态更新等]实现类,
             *   通用实现可以将support方法重写为 true, 否则需要自行判断逻辑进行实现，
             *   但切记不要暴露出异常，以免导致抛出异常而放弃其他实现类直接失败
             */
            // 根据消息类型不同 适配不同解析引擎 使用parallelStream并发执行来预防同步执行时某个方法耗时较久影响效率的问题
            messageResolves.parallelStream().forEach(resolve -> {
                // 满足support 则进行执行
                if (resolve.support(msgType)) {
                    try {
                        resolve.resolve(snCode, deviceInfo, msg);
                    } catch (Exception e) {
                        log.error("resolve: {} 执行失败", resolve.getClass(), e);
                    }
                }
            });
        } catch (Exception e) {
            log.error("主题:"+topic+" 【订阅消息解析失败】", e);
        }

    }

    /**
     * WRw6J5orBTmYzMDE4UPainH82LsI9e/h9JqGp8v1CGrFdNY6z85lSq0cxqbgZwPuVXcCN7OPxdk2LN1dP3ehCcw2zQvLJUXkgcAjymu6LWX5AK9FjHzaUZbwfNpbtlchDV+ruwjmzOOZY5JDU3/vC2C8nC3Wyvu6O/MNh+pr8oMTion7feVolkhyJyDnlswJNbTiYGb2C5opXnxQTVz5S3KZZCYISPYmPG/PJjTYkE7X0tEFpExxq57s1KCUd2rHa5cT6tsIvJLdBbn327wm0fdkCYjQZ2fd/D7cDjQUcapO03uLdnYWvecsJcon/C6kHW6/3NgzCn99PFMQkjSnNzTy+tmtsb0uxkiiKq6dqTziaxQXY2fEQ3wixWqK/VjpHnppo80drT+FDR20uBvnzna9sard2BaEbtO5F04SK2U8x/lr3pOXtxm0sh/cl2xhbqRXwYSrF1sZslmO6QuYjoYVIGOppXF/iy4Vp6aonlGHcdSdQvfZwKJVymx15XnwgyiZrheii7iOlsm8mHPKk304r4sBzxnK2luF3YYOCfknXzpnmveIcEvS3lynA2o2yI6PqfMqye6NT1O/AyzNudL5Mz92Zuk6xFRDJ/Z8atJLG0VZRL5h8HYjQvrWSBElQYjkc8Bq75xDWPHhjEx1ECcpODu2FWPSHCvXw2Fs5Jbx6cVKG9q0diqfawpY6gI+Z9Mcjc+zhIyRWuafhKgC1nKZZCYISPYmPG/PJjTYkE7CWBJAGRhJfUxi4V2vabvlct6BgcYgFWJZ+1YXMUXlAvdkCYjQZ2fd/D7cDjQUcaqgVLRhvX11oxiVU1KNx2VTP3z3mPhY7FnsE935Td5gs6+ttSHq2o9oI2NQVL7B4fp3HHpo3yF7XFkWvkgiWpV4T/lf5ZGjgH/EmBJV35l/sRN6LZ14fXbUYhdMctl0gYGnELSt40r2X+Cyseo29I51E03JuCO/upfHck80yOzYuKvpIODdbhynknaIwC79MlPQuQRAWIM07zCFHeFfpmokfTqlk3O/NpwT05uYPPlse11aRUOneYMwDYkrJH2vVS2SUGYkAluPMIW4OOhNwEkB+JTj1ZYaw5FHjRVKTrXCKCBj0KuVC70aPOFs2x76jfvJ5D1dARYk69pDMWmFjZz9mOq2Q4keM60HhGMprPyJ4Xcl/MJAFud1DY+AnwxpSzKwC9NBRlHHs9N46BUD6ZcxM+3HCYcGl6OLISekT3NupmCfB3hWgapxJ28L9bBivmjS+TM/dmbpOsRUQyf2fGrSeJ4AJCTMZ5YuFcnVccvZWcw2zQvLJUXkgcAjymu6LWV//blnDlnJU0iAUZ4KiXPXYjXvvGLW5U9stUSMJHQZZVphXXIRt2oMYkiuDY8tL/yo1EkELhM/J4zJRXXjoipbzCk81zvtbo9SLTz4qMfvDXMqMd18tKQmgCCegU1ueMwc4vpLyVmoJicajw6/zwG8QIo/iRYlLBe3bRGLBdZ+hsubIco9RwK7O5ydY2WuJIN4WpDXo1Ap57f+yT9tRYWftEu4jSr6Cm4XqX2rXoxnNvOS5StOtQk1M8vjZF0un7Hun3rJgC5waYj9Us7QJF6fz1e29XtxSKd497UXJGpSgOyX/KIyVOjeTQ3Y6EIPVVVhm76p/sZP+kZbhqyANCN9E5PaMraCoZhccb40spjkZabPwOSWED5va17Jz1pBHO1dJ2rCaEHNEXyc02hDElxQdv9mRIMysh7YjQ/JQ8Dn1A==
     */
    @Resource
    private IBzDataService bzDataService;
    private void dealBzData(String data) throws Exception {
        log.info("泵站推送原始消息：{}",data);
        //String json = data.trim();
        String json = SecretTool.decrypt(data.trim());
        log.info(json);
        JSONObject jsonObject = JSONUtil.parseObj(json.substring(1, json.length() - 1).replace("\\",""));
        BzDataEntity bzDataEntity = new BzDataEntity();
        bzDataEntity.setIp(jsonObject.getStr("BZ_IP"));
        bzDataEntity.setIpName(jsonObject.getStr("BZ_NAME"));
        bzDataEntity.setParameterConnect(jsonObject.getStr("parameterConnect"));
        bzDataEntity.setHasYewei("0");
        JSONArray array = jsonObject.getJSONArray("data");
        for (Object value : array) {
            JSONObject o = (JSONObject) value;
            for (String key : o.keySet()) {
                if (key.contains("YS_YW") || key.contains("WS_YW")) {
                    bzDataEntity.setHasYewei("1");
                    break;
                }
            }
        }
        JSONArray jsonArray = new JSONArray();
        array.forEach(object -> {
            JSONObject o = (JSONObject) object;
            JSONObject item = new JSONObject();
            o.keySet().forEach(key->{
                if(!"parameterName".equals(key) && !key.contains(".")){
                    item.set("parameterKey", key);
                    item.set("parameterValue", o.get(key));
                }else if("parameterName".equals(key)){
                    item.set("parameterName", o.get(key));
                }
            });
            jsonArray.add(item);
        });
        bzDataEntity.setJsonData(jsonArray);
        bzDataEntity.setCreateTime(LocalDateTime.now());
        bzDataService.save(bzDataEntity);
        //启停判断
        checkStartOrStop(bzDataEntity);
    }
    @Resource
    private RedisService redisCache;
    @Resource
    private IBzStartStopService bzStartStopService;
    private void checkStartOrStop(BzDataEntity bzDataEntity) {
        log.info("开始启停判断");
        String ip = bzDataEntity.getIp();
        String ipName = bzDataEntity.getIpName();
        BzDataEntity bzDataEntityRedis = redisCache.getCacheObject("pump_state_cache:"+ip);
        if(bzDataEntityRedis!=null) {
            //老数据
            JSONArray redisArray = bzDataEntityRedis.getJsonData();
            System.err.println("老数据："+redisArray);
            //新数据
            JSONArray jsonArray = bzDataEntity.getJsonData();
            System.err.println("新数据："+jsonArray);
            jsonArray.forEach(object -> {
                JSONObject o = (JSONObject) object;
                String parameterName = o.getStr("parameterName");
                if(parameterName.contains("泵状态")){
                    int parameterValue = Integer.parseInt(o.getStr("parameterValue"));
                    JSONObject jsonObject =  redisArray.stream()
                            .map(item -> (JSONObject) item).filter(item->
                         item.getStr("parameterName").equals(parameterName)
                    ).findFirst()
                            .orElse(null);
                    System.err.println("jsonObject:"+jsonObject);
                    if(jsonObject!=null){
                        System.err.println("parameterValue:"+parameterValue);
                        System.err.println("jsonObject.getInt(\"parameterValue\"):"+jsonObject.getInt("parameterValue"));
                        if(parameterValue == 0 && jsonObject.getInt("parameterValue")==1){
                            //停泵
                            BzStartStopEntity bzStartStopEntity = BzStartStopEntity.builder()
                                    .ip(ip)
                                    .ipName(ipName)
                                    .startStop("stop")
                                    .createTime(LocalDateTime.now())
                                    .pumpName(parameterName)
                                    .build();
                            bzStartStopService.save(bzStartStopEntity);
                        }
                        if(parameterValue == 1 && jsonObject.getInt("parameterValue")==0){
                            //启泵
                            BzStartStopEntity bzStartStopEntity = BzStartStopEntity.builder()
                                    .ip(ip)
                                    .ipName(ipName)
                                    .startStop("start")
                                    .createTime(LocalDateTime.now())
                                    .pumpName(parameterName)
                                    .build();
                            bzStartStopService.save(bzStartStopEntity);
                        }
                    }
                }
            });
        }
        redisCache.setCacheObject("pump_state_cache:"+ip,bzDataEntity);
    }

    private JSONObject revertToHxwl(String topic, String payload) {
        JSONObject jsonObject = JSONUtil.parseObj(payload);
        String snCode = topic.split("/")[1];
        OptDeviceDto deviceInfo = iOptDeviceService.getDeviceByRealCode(snCode);
        String deviceType = deviceInfo.getDeviceType();
        JSONObject result = new JSONObject();
        result.set("status","success");
        result.set("snCode", snCode);
        result.set("msgType", 2);
        JSONArray jsonArray = new JSONArray();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        JSONObject data = jsonObject.getJSONObject("data");
        String date = sdf.format(new Date(data.getLong("timestamp")*1000));
        if("SY800-L-Y0".equals(deviceType)) {
            JSONObject jsonYewei = new JSONObject();
            jsonYewei.set("monitorTime", date);
            jsonYewei.set("dataUnit", "m");
            jsonYewei.set("monitorData", data.getDouble("yewei")+"");
            jsonYewei.set("monitorName", "液位");
            jsonYewei.set("monitorCode", "0x05D1");
            jsonArray.add(jsonYewei);
            JSONObject jsonSpeed = new JSONObject();
            jsonSpeed.set("monitorTime", date);
            jsonSpeed.set("dataUnit", "m/s");
            jsonSpeed.set("monitorData", data.getDouble("speed")+"");
            jsonSpeed.set("monitorName", "流速");
            jsonSpeed.set("monitorCode", "0x0593");
            jsonArray.add(jsonSpeed);
            JSONObject jsonTemperature = new JSONObject();
            jsonTemperature.set("monitorTime", date);
            jsonTemperature.set("dataUnit", "℃");
            jsonTemperature.set("monitorData", data.getDouble("temperature")+"");
            jsonTemperature.set("monitorName", "温度");
            jsonTemperature.set("monitorCode", "0x0502");
            jsonArray.add(jsonTemperature);
            JSONObject jsonFlow = new JSONObject();
            jsonFlow.set("monitorTime", date);
            jsonFlow.set("dataUnit", "m³/s");
            jsonFlow.set("monitorData", data.getDouble("flow")+"");
            jsonFlow.set("monitorName", "瞬时流速");
            jsonFlow.set("monitorCode", "0x0592");
            jsonArray.add(jsonFlow);
            JSONObject jsonTotalFlow = new JSONObject();
            jsonTotalFlow.set("monitorTime", date);
            jsonTotalFlow.set("dataUnit", "m³");
            jsonTotalFlow.set("monitorData", data.getDouble("totalflow")+"");
            jsonTotalFlow.set("monitorName", "累计流量");
            jsonTotalFlow.set("monitorCode", "0x0590");
            jsonArray.add(jsonTotalFlow);
        }else if("SR500-L-Y0-SNY".equals(deviceType)) {
            JSONObject jsonCod = new JSONObject();
            jsonCod.set("monitorTime", date);
            jsonCod.set("dataUnit", "mg/L");
            jsonCod.set("monitorData", data.getDouble("cod") + "");
            jsonCod.set("monitorName", "COD");
            jsonCod.set("monitorCode", "0x0527");
            jsonArray.add(jsonCod);
            JSONObject jsonNtu = new JSONObject();
            jsonNtu.set("monitorTime", date);
            jsonNtu.set("dataUnit", "NTU");
            jsonNtu.set("monitorData", data.getDouble("NTU") + "");
            jsonNtu.set("monitorName", "浊度");
            jsonNtu.set("monitorCode", "0x0520");
            jsonArray.add(jsonNtu);
            JSONObject jsonNh3 = new JSONObject();
            jsonNh3.set("monitorTime", date);
            jsonNh3.set("dataUnit", "mg/L");
            jsonNh3.set("monitorData", data.getDouble("nh3") + "");
            jsonNh3.set("monitorName", "氨氮");
            jsonNh3.set("monitorCode", "0x0528");
            jsonArray.add(jsonNh3);
            JSONObject jsonTemperature = new JSONObject();
            jsonTemperature.set("monitorTime", date);
            jsonTemperature.set("dataUnit", "℃");
            jsonTemperature.set("monitorData", data.getDouble("nh3_temperature") + "");
            jsonTemperature.set("monitorName", "水温");
            jsonTemperature.set("monitorCode", "0x0502");
            jsonArray.add(jsonTemperature);
        }else if("SR500-N-D2-LD".equals(deviceType)) {
            JSONObject jsonWaterLevel = new JSONObject();
            jsonWaterLevel.set("monitorTime", date);
            jsonWaterLevel.set("dataUnit", "m");
            jsonWaterLevel.set("monitorData", data.getDouble("yewei") + "");
            jsonWaterLevel.set("monitorName", "液位");
            jsonWaterLevel.set("monitorCode", "0x05D1");
            jsonArray.add(jsonWaterLevel);
            JSONObject jsonVol = new JSONObject();
            jsonVol.set("monitorTime", date);
            jsonVol.set("dataUnit", "V");
            jsonVol.set("monitorData", data.getDouble("battery_voltage") + "");
            jsonVol.set("monitorName", "电压");
            jsonVol.set("monitorCode", "0x0501");
            jsonArray.add(jsonVol);
            JSONObject jsonSoc = new JSONObject();
            jsonSoc.set("monitorTime", date);
            jsonSoc.set("dataUnit", "%");
            jsonSoc.set("monitorData", data.getDouble("battery") + "");
            jsonSoc.set("monitorName", "电量百分比");
            jsonSoc.set("monitorCode", "0x6500");
            jsonArray.add(jsonSoc);
        }else if("SE200-L-Y0".equals(deviceType)) {
            JSONObject jsonTodayRain = new JSONObject();
            jsonTodayRain.set("monitorTime", date);
            jsonTodayRain.set("dataUnit", "mm");
            jsonTodayRain.set("monitorData", data.getDouble("today_rain") + "");
            jsonTodayRain.set("monitorName", "当天降雨量");
            jsonTodayRain.set("monitorCode", "0x0567");
            jsonArray.add(jsonTodayRain);
            JSONObject jsonRain = new JSONObject();
            jsonRain.set("monitorTime", date);
            jsonRain.set("dataUnit", "mm");
            jsonRain.set("monitorData", data.getDouble("rain") + "");
            jsonRain.set("monitorName", "瞬时降雨量");
            jsonRain.set("monitorCode", "0x0568");
            jsonArray.add(jsonRain);
            JSONObject jsonYesterdayRain = new JSONObject();
            jsonYesterdayRain.set("monitorTime", date);
            jsonYesterdayRain.set("dataUnit", "mm");
            jsonYesterdayRain.set("monitorData", data.getDouble("yesterday_rain") + "");
            jsonYesterdayRain.set("monitorName", "昨日降雨量");
            jsonYesterdayRain.set("monitorCode", "0x0566");
            jsonArray.add(jsonYesterdayRain);
            JSONObject jsonTotalRain = new JSONObject();
            jsonTotalRain.set("monitorTime", date);
            jsonTotalRain.set("dataUnit", "mm");
            jsonTotalRain.set("monitorData", data.getDouble("total_rain") + "");
            jsonTotalRain.set("monitorName", "总降雨量");
            jsonTotalRain.set("monitorCode", "0x0569");
            jsonArray.add(jsonTotalRain);
            JSONObject jsonHourRain = new JSONObject();
            jsonHourRain.set("monitorTime", date);
            jsonHourRain.set("dataUnit", "mm");
            jsonHourRain.set("monitorData", data.getDouble("hour_rain") + "");
            jsonHourRain.set("monitorName", "小时降雨量");
            jsonHourRain.set("monitorCode", "0x056A");
            jsonArray.add(jsonHourRain);
            JSONObject jsonLastHourRain = new JSONObject();
            jsonLastHourRain.set("monitorTime", date);
            jsonLastHourRain.set("dataUnit", "mm");
            jsonLastHourRain.set("monitorData", data.getDouble("last_hour_rain") + "");
            jsonLastHourRain.set("monitorName", "上小时降雨量");
            jsonLastHourRain.set("monitorCode", "0x056B");
            jsonArray.add(jsonLastHourRain);
            JSONObject jsonSoc = new JSONObject();
            jsonSoc.set("monitorTime", date);
            jsonSoc.set("dataUnit", "mm");
            jsonSoc.set("monitorData", data.getDouble("soc") + "");
            jsonSoc.set("monitorName", "电量百分比");
            jsonSoc.set("monitorCode", "0x6500");
            jsonArray.add(jsonSoc);
        }else{

        }
        result.set("msg", jsonArray);
        return result;
    }
}
